package com.atguigu.edu.realtime.app.dwd.db;

import com.atguigu.edu.realtime.util.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Author: Felix
 * Date: 2022/5/12
 * Desc: 互动域评价事务事实表
 */
public class DwdInteractionReview {
    public static void main(String[] args) throws Exception {

        // TODO 1. 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO 2. 检查点相关设置(略)

        // TODO 3. 从 Kafka 读取 topic_db 数据，封装为 Flink SQL 表
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("dwd_interaction_review"));


        // TODO 4. 读取评论表数据
        Table reviewInfo = tableEnv.sqlQuery("select " +
                "data['id'] id, " +
                "data['user_id'] user_id, " +
                "data['course_id'] course_id, " +
                "data['review_stars'] review_stars, " +
                "data['create_time'] create_time, " +
                "data['review_txt'] `review`, " +
                "proc_time, " +
                "ts " +
                "from topic_db " +
                "where `table` = 'review_info' " +
                "and `type` = 'insert' ");
        tableEnv.createTemporaryView("review_info", reviewInfo);


        // TODO 6. 关联两张表
        Table resultTable = tableEnv.sqlQuery("select " +
                "id, " +
                "user_id, " +
                "course_id, " +
                "review_stars, " +
                "date_format(create_time,'yyyy-MM-dd') date_id, " +
                "create_time, " +
                "`review`, " +
                "ts " +
                "from review_info  " );

        tableEnv.createTemporaryView("result_table", resultTable);

        // TODO 7. 建立 Upsert-Kafka dwd_interaction_review 表
        tableEnv.executeSql("create table dwd_interaction_review( " +
                "id string, " +
                "user_id string, " +
                "course_id string, " +
                "review_stars string, " +
                "date_id string, " +
                "create_time string, " +
                "`review` string, " +
                "ts string, " +
                "primary key(id) not enforced " +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_interaction_review"));

        // TODO 8. 将关联结果写入 Upsert-Kafka 表
        tableEnv.executeSql("" +
                "insert into dwd_interaction_review select * from result_table");
    }

}
